package com.example.magical.normal.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.BatchReceivePolicy;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.Messages;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.client.api.SubscriptionType;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.concurrent.TimeUnit;

@Component
@Slf4j
public class PulsarPayhubRun {
    @Value("${PULSAR_SERVER_URL_PAYHUB}")
    String SERVER_URL;
    @Value("${PULSAR_TOPIC_PAYHUB}")
    String TOPIC;
    @Value("${PULSAR_TOKEN_PAYHUB}")
    String TOKEN;

    @PostConstruct
    public void run() {
        ClientPulsar clientPulsar = new ClientPulsar(SERVER_URL, TOKEN);
        try (Consumer<byte[]> consumer = clientPulsar.getClient().newConsumer()
                .topic(TOPIC)
                .subscriptionName("payhub-1")
                .subscriptionType(SubscriptionType.Shared)
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                .batchReceivePolicy(new BatchReceivePolicy.Builder()
                        .maxNumMessages(2000)
                        .maxNumBytes(0)
                        .timeout(10, TimeUnit.SECONDS)
                        .build())
                .receiverQueueSize(2000)
                .subscribe()) {
            log.info("开始处理PAYHUB消息");
            while (true) {
                Messages<byte[]> messages = consumer.batchReceive();
                if (messages.size() > 0) {
                    for (Message<byte[]> message : messages) {
                        String data = new String(message.getData());
                        log.info("message:{}", data);
                    }
                }
            }
        } catch (PulsarClientException e) {
            log.error("接收消息失败：", e);
        }
    }
}
